package snippets.cse524.activeDates;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import edu.umd.cloud9.collection.wikipedia.WikipediaPage;
import edu.umd.cloud9.collection.wikipedia.WikipediaPageInputFormat;
import edu.umd.cloud9.io.array.ArrayListWritable;
import edu.umd.cloud9.io.array.ArrayListWritableComparable;
import edu.umd.cloud9.io.pair.PairOfInts;
import edu.umd.cloud9.io.pair.PairOfStringInt;
import edu.umd.cloud9.io.pair.PairOfStrings;
import edu.umd.cloud9.io.pair.PairOfWritables;

public class GenderEvaluatorMR extends Configured implements Tool {

	public static class Map extends MapReduceBase implements
			Mapper<LongWritable, WikipediaPage, Text, Text> {
		private static final IntWritable ONE = new IntWritable(1);
		public int[] GAPS = { 1, 5, 10, 15, 20 };
		private static final Pattern male = Pattern
				.compile("([^\\w]|^)(his|he|him)([^\\w]|$)");
		private static final Pattern female = Pattern
				.compile("([^\\w]|^)(her|she)([^\\w]|$)");

		@Override
		public void map(LongWritable key, WikipediaPage page,
				OutputCollector<Text, Text> collector, Reporter arg3)
				throws IOException {
			if (!page.isArticle()) {
				return;
			}

			String content = null;

			try {
				content = page.getWikiMarkup();
			} catch (Exception e) {
				return;
			}

			WikipediaDateExtractor wde = new WikipediaDateExtractor(content, page.getTitle());

			if (!wde.isEntity()) {
				return;
			}

			int limit = 3000;
			limit = limit > content.length() ? content.length() : limit;
			content = content.substring(0, limit);

			Matcher femaleMatcher = female.matcher(content);
			Matcher maleMatcher = male.matcher(content);

			int femaleCount = 0, maleCount = 0;

			while (femaleMatcher.find())
				++femaleCount;
			while (maleMatcher.find())
				++maleCount;
			
			String normalizedTitle = StringUtils.replaceChars(page.getTitle(), ' ', '_');

			collector.collect(
					new Text(normalizedTitle),
					new Text(femaleCount + ":" + maleCount + "\t" + (femaleCount > maleCount ? "female" : ( maleCount > femaleCount ? "male" : "unknown"))));
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new GenderEvaluatorMR(), args);

	}

	@Override
	public int run(String[] args) throws Exception {
		String inputPath = args[0];
		String outputPath = args[1];

		JobConf conf = new JobConf(getConf(), GenderEvaluatorMR.class);
		conf.setJobName(String.format("GenderEvaluatorMR [%s: %s]", inputPath,
				outputPath));

		conf.setNumMapTasks(12);
		conf.setNumReduceTasks(12);

		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));

		conf.setInputFormat(WikipediaPageInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		conf.setMapperClass(Map.class);

		// Set the outputs for the Map
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>> mapOpDummy = new
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>>();
		Text mapOpDummy = new Text();

		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(mapOpDummy.getClass());

		JobClient.runJob(conf);

		return 0;
	}

}
